package com.chukun.flink.dataset.operator;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @author chukun
 * @version 1.0.0
 * @description reduceGroup使用GroupCombineFunction函数进行预先局部计算
 *              可预计算的ReduceGroup
 * @createTime 2022年05月28日 20:35:00
 */
public class ReduceGroupCombineOperator {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        List<Tuple2<String, Integer>> list = new ArrayList<>();
        list.add(new Tuple2<>("张三", 15));
        list.add(new Tuple2<>("张三", 20));
        list.add(new Tuple2<>("张三", 21));
        list.add(new Tuple2<>("李四", 31));
        list.add(new Tuple2<>("李四", 38));
        list.add(new Tuple2<>("李四", 45));
        list.add(new Tuple2<>("王五", 55));
        list.add(new Tuple2<>("王五", 67));

        DataSet<Tuple2<String, Integer>> sourceDataSet = env.fromCollection(list);

        // 调用 CustomCombinableGroupReducer 接口，包含GroupReduceFunction与GroupCombineFunction
        DataSet<Tuple2<String, Integer>> groupReduceCombineDataSet = sourceDataSet.groupBy((key) -> key.f0)
                .reduceGroup(new CustomCombinableGroupReducer());

        //显式调用 combineGroup
        DataSet<Tuple2<String, Integer>> groupReduceCallCombineDataSet = sourceDataSet.groupBy((key) -> key.f0)
                .combineGroup(new CustomCombinableGroupReducer())
                .groupBy((key) -> key.f0)
                .reduceGroup(new CustomCombinableGroupReducer());

        groupReduceCombineDataSet.print("GroupReduceFunction--->GroupCombineFunction");

        groupReduceCallCombineDataSet.print("Call -->GroupReduceFunction--->GroupCombineFunction");

        env.execute("ReduceGroupCombineOperator");
    }

    /**
     * 局部计算可以显著提高reduceGroup操作符的性能，使用GroupCombineFunction，使reduceGroup函数具备局部计算的能力
     */
    private static class CustomCombinableGroupReducer implements
            GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>,
            GroupCombineFunction<Tuple2<String,Integer>, Tuple2<String, Integer>> {

        // 先被调用
        @Override
        public void combine(Iterable<Tuple2<String, Integer>> values, Collector<Tuple2<String, Integer>> out) throws Exception {
            dealCompute(values, out, "调用combine方法进行最终计算");
        }

        // 后被调用
        @Override
        public void reduce(Iterable<Tuple2<String, Integer>> values, Collector<Tuple2<String, Integer>> out) throws Exception {
            dealCompute(values, out, "调用reduce方法进行最终计算");
        }

        private void dealCompute(Iterable<Tuple2<String, Integer>> values, Collector<Tuple2<String, Integer>> out, String flag) {
            String key = "";
            int value = 0;
            String str = "";
            Iterator<Tuple2<String, Integer>> iterator = values.iterator();
            while (iterator.hasNext()) {
                Tuple2<String, Integer> next = iterator.next();
                str = StringUtils.join(str, next, " ");
                key = next.f0;
                value = value + next.f1;
            }
            str = StringUtils.join(flag, str);
            System.out.println(str);
            out.collect(new Tuple2<>(key, value));
        }
    }
}
